package job;

import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;
import kafka.serializer.StringDecoder;
import play.Logger;
import play.jobs.Every;
import play.jobs.Job;
import utils.DateUtils;
import utils.FNVHash;
import utils.MsgsendlogUtil;
import utils.PropertyUtil;
import utils.SystemParam.Topic;
import utils.TopicUtil;
import bp.NetSaleBP;

/**
 * 
 * <b>类描述：错误消息消费者</b><br/>
 * <b>类名称：</b>ErrorMessageConsumer<br/>
 * <b>创建人：</b>郭子义</a><br/>
 * <b>关键修改：</b><br/>
 * <b>修改时间：</b><br/>
 * <b>修改人：</b><br/>
 * 
 */
@SuppressWarnings("rawtypes")
@Every("3s")
public class ErrorMessageConsumer extends Job {
	private final ConsumerConnector consumer;
	private final String topic;
	private static Map<String,Integer> topicOnlines = new HashMap<String,Integer>();

	public ErrorMessageConsumer() {

		System.out.println("before create consumer ErrorConsumer");
		consumer = kafka.consumer.Consumer.createJavaConsumerConnector(createConsumerConfig());// 创建到zookeeper的连接
		System.out.println("after create consumer ErrorConsumer");
		this.topic = TopicUtil.getTopic("ErrorConsumer");
	}

	private ConsumerConfig createConsumerConfig() {
		Properties props = new Properties();
		props.put("zookeeper.connect", KafkaProperties.zkConnect);// 要连接的zookeeper地址
		props.put("group.id", KafkaProperties.groupId);// 消费者所在的组
	//	props.put("consumer.id", "ErrorConsumer");// 此消费者的唯一标识用于在zookeeper中存储此消费者的偏移量
		props.put("zookeeper.session.timeout.ms", KafkaProperties.zSessionTimeout);// zookeeper连接超时时间
		//props.put("zookeeper.sync.time.ms", "40000");
		props.put("auto.commit.enable", "false");
		props.put("auto.offset.reset", "smallest");

		return new ConsumerConfig(props);
	}

	@Override
	public void doJob() throws Exception {
		int maxconsumercount = Integer.parseInt(PropertyUtil.getProperty("maxconsumercount"));
		//isonline是否重启，当BUG解决后重启服务isonline为0，重启并且最大重复消费次数为0时，直接返回
//		if (isonline == 1 && maxconsumercount == 0) {
//			return;
//		}

		Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
		topicCountMap.put(topic, new Integer(1));// topic:要取的主题，Integer:创建的数据流个数
		Map<String, List<KafkaStream<String, String>>> consumerMap = consumer.createMessageStreams(topicCountMap, new StringDecoder(null), new StringDecoder(null));// 获取创建的流
		List<KafkaStream<String, String>> list = consumerMap.get(topic);// 取得某一个消息的流
		KafkaStream<String, String> kafkaStream = list.get(0);// 刚才创建流是传入的流个数是1，所以通过get(0)获取
		ConsumerIterator<String, String> it = kafkaStream.iterator();// 遍历消息，此函数会一直阻塞
		MessageAndMetadata<String, String> data = null;
		String delaytopic = TopicUtil.getTopic("SaleDelay");
		while (it.hasNext()) {
			data = it.next();
			// 此处做对消息的处理
			boolean flag = false;
			//错误消息
			String message = data.message();
			if(message.indexOf(delaytopic) >= 0){
				consumer.commitOffsets();
				continue;
			}
			int beginindex = 0;
			//错误消息主题
			int endindex = message.indexOf("#");
			String topic = message.substring(beginindex,endindex);
			//获取此主题BUG是否解决并重启
			Integer isonline = topicOnlines.get(topic);
			if(isonline == null){
				isonline = 0;
				topicOnlines.put(topic, isonline);
			}
			//错误消息消费次数
			beginindex = endindex+1;
			endindex = message.indexOf("#", beginindex);
			int count = Integer.parseInt(message.substring(beginindex,endindex));
			SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
			//错误消息上次消费时间加3分钟
			beginindex = endindex+1;
			endindex = message.indexOf("#", beginindex);
			Date datetime = DateUtils.add(sdf.parse(message.substring(beginindex,endindex)), Calendar.MINUTE, 3); 
			Date nowtime = new Date();
			//错误消息主体
			beginindex = endindex+1;
			String errormessage = message.substring(beginindex);
			//当消费次数大于等于最大消费次数时，如果BUG解决重启后，继续消费。没有则直接重发
			if (count >= maxconsumercount) {
				if (isonline == 0) {
					invoke(flag, topic, count, errormessage);
				} else {
					//send(TopicUtil.getTopic("SaleDelay"), message);
					Logger.log4j.error("该消息超过最大消费次数，不再消费，主题："+topic+",消息："+errormessage);
				}
			} else {
				//当上次消费时间加3分钟小于当前时间时，消费消息，否则重新发送
				if (datetime.before(nowtime)) {
					invoke(flag, topic, count, errormessage);
				} else {
					send(TopicUtil.getTopic("SaleDelay"), message);
				}
			}
			consumer.commitOffsets();// 提交此消费者的偏移量
			//break;
		}
	}
	
	public static void send(String topic, String message){
		boolean issendsuccess = true;
		try{
			ProducerFactory.send(topic, message);
		}catch(Exception e){
			Logger.log4j.error("消息发送到kafka失败 ： "+message ,e);
			issendsuccess = false;
		}
		
//		MsgsendlogUtil.write(null, message, new Date(), topic.name(), "ErrorMessage", null, null,
//				FNVHash.hash64(message), null,true,issendsuccess);
	}

	private void invoke(boolean flag, String topic, int count, String errormessage) {
		try {
			if ("LOCK".equals(topic)) {
				flag = new NetSaleBP().lock(errormessage);
			} else if ("SELL".equals(topic)) {
				flag = new NetSaleBP().sell(errormessage);
			} else if ("UNLO".equals(topic)) {
				flag = new NetSaleBP().unlock(errormessage);
			} else if ("RETR".equals(topic)) {
				flag = new NetSaleBP().returnticket(errormessage);
			} else if ("REUK".equals(topic)) {
				flag = new NetSaleBP().returnunknown(errormessage);
			} else if ("SEUK".equals(topic)) {
				flag = new NetSaleBP().sellunknown(errormessage);
			} else if ("CHTI".equals(topic)) {
				flag = new NetSaleBP().changeticketno(errormessage);
			} else if ("CACL".equals(topic)) {
				flag = new NetSaleBP().cancelticket(errormessage);
			} else if ("CLUK".equals(topic)) {
				flag = new NetSaleBP().cancelunknown(errormessage);
			} else if ("REIN".equals(topic)) {
				flag = new NetSaleBP().cancelinsurance(errormessage);
			}
		} catch (Exception e) {
		}

		//消费失败，更改isonline为1，增加消费次数，并发送错误消息
		if (!flag) {
			topicOnlines.put(topic, 1);
			count++;
			ProducerFactory.send(TopicUtil.getTopic("SaleDelay"), topic + "#" + count + "#"+DateUtils.formatDatetime(new Date())+"#" + errormessage);
		}
	}
	
	public static void setOnline(String topic ){
		topicOnlines.put(topic, 1);
	}

	
}
